﻿using System;
using System.Collections.Generic;
using PhiStore.Communication;
using PhiStore.Packages;
using PhiStore.Packages.InternalPackages;
using PhiStore.Policies;
using PhiStore.Storage;
using PhiStore.Threading;
using PhiStore.Util;

namespace PhiStore.ObjectManaging
{
    class ObjectManager : Boostable
    {
        private Dictionary<string, ObjectEntry> objects;

        private Channel<PhiInternalPackage> actionChannel;
        public ObjectManager(Channel<PhiInternalPackage> actionChannel)
        {
            Log.Instance.AddLogEntry("ObjectManager", "Initializing Object Manager");

            this.actionChannel = actionChannel;
            StorageManager.Instance.RegisterObjectManager(actionChannel);
            objects = new Dictionary<string, ObjectEntry>();
        }

        List<ObjectDevice> objectDevices = new List<ObjectDevice>();
        ObjectDevice Fastest { get { return objectDevices[objectDevices.Count - 1]; } }

        public void Run()
        {
            try
            {
                Log.Instance.AddLogEntry("ObjectManager", "Object Manager Starting");
                while (true)
                {
                    PhiInternalPackage action = actionChannel.Receive();
                    Log.Instance.AddLogEntry("ObjectManager", "received package: " + action.ToString());

                    if (action is DeviceAdded)
                    {
                        addObjectsForDevice((action as DeviceAdded).Device);
                    }
                    else if (action is ObjectCreate)
                    {
                        createObject((ObjectCreate)action);
                    }
                    else if (action is NegotiationInitialization)
                    {
                        InitializeNegotiation((NegotiationInitialization)action);
                    }
                    else if (action is NegotiationSuccess)
                    {
                        SetThisNegotiationWinner((NegotiationSuccess)action);
                    }
                    else if (action is NegotiationWinner)
                    {
                        SetNegotiationWinner((NegotiationWinner)action);
                    }
                    else if (action is ExternalNegotiation)
                    {
                        CheckNegotiation((ExternalNegotiation)action);
                    }
                    else if (action is LockObjectEntry)
                    {
                        LockObject((LockObjectEntry)action);
                    }
                    else if (action is UnlockObjectEntry)
                    {
                        UnlockObject((UnlockObjectEntry)action);
                    }
                    else if (action is RemoveObject)
                    {
                        RemoveObject((RemoveObject)action);
                    }
                    else if (action is InternalListObjects)
                    {
                        ListObjects((InternalListObjects)action);
                    }
                    else if (action is AddRemoteObject)
                    {
                        AddObject((AddRemoteObject)action);
                    }
                    else if (action is InternalGetStats)
                    {
                        ReturnStats((InternalGetStats)action);
                    }
                    else if (action is InternalGetPolicy)
                    {
                        ReturnPolicy((InternalGetPolicy)action);
                    }
                    else if (action is GarbageCollect)
                    {
                        Log.Instance.AddLogEntry("ObjectManager", "Running garbage collection");
                        GarbageCollect();
                    }
                    else if (action is StopGarbageCollect)
                    {
                        Log.Instance.AddLogEntry("ObjectManager", "Garbage collection stopped");
                    }
                    else
                    {
                        Log.Instance.AddLogEntry("ObjectManager", action.ToString() + " is an unsupported internal package - plx fix!");
                    }
                }

            }
            catch (PoisonException)
            {
                Log.Instance.AddLogEntry("ObjectManager", "Shutting down");
                //bye bye
            }
            CleanUpWaiting();
        }

        private void CleanUpWaiting()
        {
            foreach (ObjectEntry oe in objects.Values)
            {
                foreach (Channel<object> channel in oe.WaitingProcesses)
                {
                    channel.Poison();
                }
            }
        }

        private void ReturnPolicy(InternalGetPolicy package)
        {
            Log.Instance.AddLogEntry("ObjectManager", package.Key);
            if (objects.ContainsKey(package.Key))
            {
                Policy policy = objects[package.Key].ObjectLock.Policy;
                package.ReplyChannel.Send(policy);
            }
            else
                package.ReplyChannel.Send(ReturnCode.ObjectNotFound);
        }

        private void GarbageCollect()
        {
            //perform garbagecollection until a new package is received

            bool allFull = true;
            foreach (ObjectDevice dev in objectDevices)
            {
                if (actionChannel.Count > 0) return; //STOP

                dev.Benchmark();
                allFull &= dev.HighWaterMarkReached;
            }

            if (allFull)
            {
                int loop = 0;
                while (loop++ < 20)
                    Log.Instance.AddLogEntry("ObjectManager", "All Devices are full - shutting down");
                PhiServer.Instance.Stop();
            }

            objectDevices.Sort(); //restructuring device-list

            if (actionChannel.Count > 0) return; //STOP

            string[] objectNames = ArrayUtil.CopyTo<string, ICollection<string>>(objects.Keys);

            if (actionChannel.Count > 0) return; //STOP

            foreach (string objectName in objectNames)
            {
                ObjectEntry oe = objects[objectName];

                if (actionChannel.Count > 0) return;//STOP

                TimeSpan ts = TimeSpan.FromMilliseconds(Convert.ToDouble(oe.ObjectLock.Policy.IdleThreshold));
                DateTime lastAccess = oe.ObjectLock.Device.LatestAccess(objectName);

                if (actionChannel.Count > 0) return;//STOP


                if (lastAccess.Add(ts) < DateTime.Now) //demote / remove
                {
                    if (oe.ObjectLock.Policy.MayGarbageCollect)
                    {
                        oe.ObjectLock.Device.DeleteObject(objectName, false);
                        objects.Remove(objectName);
                        Log.Instance.AddLogEntry("ObjectManager", "GC: " + objectName + " removed");

                    }
                    else if (objectDevices.Count > 1) //more than one device
                    {
                        //try to move to a slower device
                        int devicerank = objectDevices.IndexOf(oe.ObjectLock.Device);

                        if (devicerank != 0) //already @ slowest device
                        {

                            ObjectDevice newDevice = objectDevices[devicerank - 1];
                            if (!newDevice.HighWaterMarkReached)
                            {
                                MoveObject(objectName, newDevice);
                                Log.Instance.AddLogEntry("ObjectManager", "GC: " + objectName + " demoted to slower device");
                            }
                        }
                    }
                }
                else if (lastAccess.Add(ts) >= DateTime.Now) //promote
                {
                    if (objectDevices.Count > 1) //more than one device
                    {
                        //try to move to a slower device
                        int devicerank = objectDevices.IndexOf(oe.ObjectLock.Device);

                        if (devicerank != objectDevices.Count - 1) //already @ fastest device
                        {
                            ObjectDevice newDevice = objectDevices[devicerank + 1];
                            MoveObject(objectName, newDevice);
                            Log.Instance.AddLogEntry("ObjectManager", "GC: " + objectName + " promoted to faster device");
                        }
                    }
                }
            }

            System.GC.Collect();

            Log.Instance.AddLogEntry("ObjectManager", "GC Completed - checking for dead connections");

            PeerManager.Instance.RemovePeer(null); //this call will eventually initialize the GC again

        }

        private void MoveObject(string key, ObjectDevice newDevice)
        {
            ObjectInformation oi = objects[key].ObjectLock.ObjectInformation;
            ObjectDevice oldDevice = objects[key].ObjectLock.Device;
            newDevice.CreateObject(key, objects[key].ObjectLock.Policy, oi, false);
            for (long i = 0; i < oi.Count; i++)
            {
                long index = oi.FirstIndex + i;
                newDevice.CreateObject(key, index, oldDevice.ReadObject(key, index), false);
            }

            oldDevice.DeleteObject(key, false);
            objects[key].ObjectLock.Device = newDevice;
        }

        private void ReturnStats(InternalGetStats package)
        {
            Log.Instance.AddLogEntry("ObjectManager", package.Key);
            if (objects.ContainsKey(package.Key))
            {
                ObjectInformation oi = objects[package.Key].ObjectLock.ObjectInformation.Copy();
                package.ReplyChannel.Send(oi);
            }
            else
                package.ReplyChannel.Send(ReturnCode.ObjectNotFound);
        }

        private void AddObject(AddRemoteObject package)
        {
            string key = package.obj.key;
            Log.Instance.AddLogEntry("ObjectManager", key);

            if (!objects.ContainsKey(key))
            {
                ObjectLock ol = new ObjectLock();
                ol.Device = Fastest;
                ol.ObjectInformation = new ObjectInformation(0L, 0L, 0L); //I don't have a copy
                ol.Policy = package.obj.Policy;
                ol.Coordinator = false;

                //Create physical copies!
                ol.Device.CreateObject(key, ol.Policy, ol.ObjectInformation, false);

                ol.ExternalCoordinator = package.obj.Coordinator ? package.p : null;


                ObjectEntry oe = new ObjectEntry();
                oe.BullyNumber = 0L;
                oe.Busy = false;

                oe.Negotiating = false;
                oe.ObjectLock = ol;

                objects.Add(key, oe);
            }
            else if (package.obj.Coordinator)
            {
                objects[key].ObjectLock.ExternalCoordinator = package.p;
            }
        }

        private void ListObjects(InternalListObjects package)
        {
            List<SimpleObjectEntry> objs = new List<SimpleObjectEntry>();
            foreach (string key in objects.Keys)
            {
                SimpleObjectEntry obj = new SimpleObjectEntry();
                obj.key = key;
                obj.Policy = objects[key].ObjectLock.Policy.Copy();
                obj.Coordinator = objects[key].ObjectLock.Coordinator;
                objs.Add(obj);
            }

            package.ReplyChannel.Send(objs);
        }

        #region Negotiation

        private void CheckNegotiation(ExternalNegotiation neg)
        {
            string key = neg.Negotiation.Key;
            if (!objects.ContainsKey(key)) //this peer doesn't have object, create a dummy
            {
                Log.Instance.AddLogEntry("ObjectManager", "External negotiation received for an object which doesn't exist");
                createObject(new ObjectCreate(null, new Create(new Policy(), new List<byte[]>()), key));
            }

            if (!objects[key].Negotiating)
            {
                Log.Instance.AddLogEntry("ObjectManager", "Not negotiating any more");
                neg.ReplyChannel.Send(new NegotiationReply(ReturnCode.NegotiationLost, objects[key].ObjectLock.ObjectInformation.Version));
            }
            else if (neg.Negotiation.BullyNumber > objects[key].BullyNumber)
            {
                Log.Instance.AddLogEntry("ObjectManager", "Negotiation won in this case ");
                objects[key].BullyNumber = neg.Negotiation.BullyNumber;
                neg.ReplyChannel.Send(new NegotiationReply(ReturnCode.Success, objects[key].ObjectLock.ObjectInformation.Version));
            }
            else if (neg.Negotiation.BullyNumber == objects[key].BullyNumber)
            {
                Log.Instance.AddLogEntry("ObjectManager", "Heureka - A negotiation conflict occured... those are rare!");
                neg.ReplyChannel.Send(new NegotiationReply(ReturnCode.NegotiationConflict, objects[key].ObjectLock.ObjectInformation.Version));
            }
            else
            {
                Log.Instance.AddLogEntry("ObjectManager", "Negotiation lost in this case ");
                neg.ReplyChannel.Send(new NegotiationReply(ReturnCode.NegotiationLost, objects[key].ObjectLock.ObjectInformation.Version));
            }
        }

        private void InitializeNegotiation(NegotiationInitialization neg)
        {
            Log.Instance.AddLogEntry("ObjectManager", neg.Key);
            if (!objects.ContainsKey(neg.Key))
            {
                neg.ReplyChannel.Send(ReturnCode.ObjectNotFound);
            }
            else
            {
                ObjectEntry oe = objects[neg.Key];
                //if (oe.Busy)
                //{
                //    package.ReplyChannel.Send(ReturnCode.ObjectIsBusy);
                //}
                //else 

                // assuming that the object isn't "busy"
                if (oe.Negotiating)
                {
                    Log.Instance.AddLogEntry("ObjectManager", "Already negotiating - go away");
                    neg.ReplyChannel.Send(ReturnCode.NegotiationLost);
                }
                
                else
                {
                    oe.Busy = true;
                    oe.Negotiating = true;
                    ulong bullyNumber = RandomGenerator.ULong;
                    oe.BullyNumber = bullyNumber;
                    neg.ReplyChannel.Send(ReturnCode.Success);
                    neg.ReplyChannel.Send(bullyNumber);
                    neg.ReplyChannel.Send(oe.ObjectLock);
                }
            }
        }

        private void SetNegotiationWinner(NegotiationWinner neg)
        {
            string key = neg.Action.Key;
            Log.Instance.AddLogEntry("ObjectManager", key);
            Log.Instance.AddLogEntry("ObjectManager", "External Peer winner: " + neg.Winner.BC.ToString());

            objects[key].ObjectLock.Coordinator = false;
            objects[key].ObjectLock.ExternalCoordinator = neg.Winner;
            objects[key].ObjectLock.Device = Fastest;
            objects[key].Negotiating = false;
            objects[key].Busy = true;
            objects[key].BullyNumber = 0L;
            objects[key].ObjectLock.Policy = neg.Action.Policy;
            neg.ReplyChannel.Send(objects[key].ObjectLock);
        }

        private void SetThisNegotiationWinner(NegotiationSuccess neg)
        {
            Log.Instance.AddLogEntry("ObjectManager", neg.Key);
            objects[neg.Key].ObjectLock.Coordinator = true;
            objects[neg.Key].Busy = true;
            objects[neg.Key].ObjectLock.Device = Fastest;
            objects[neg.Key].Negotiating = false;
            objects[neg.Key].BullyNumber = 0L;

            //ObjectInformation oi = objects[neg.Key].ObjectLock.ObjectInformation;
            //objects[neg.Key].ObjectLock.ObjectInformation = new ObjectInformation(oi.Version, oi.Count, oi.FirstIndex); //version should be 1 here

            neg.replyChannel.Send(objects[neg.Key].ObjectLock);//send OL for performing create operation!
        }

        #endregion

        private void RemoveObject(RemoveObject package)
        {
            Log.Instance.AddLogEntry("ObjectManager", package.Key);
            if (objects.ContainsKey(package.Key))
            {
                if (objects[package.Key].WaitingProcesses.Count > 0)
                {
                    foreach (Channel<object> waiting in objects[package.Key].WaitingProcesses)
                    {
                        waiting.Send(ReturnCode.ObjectNotFound);
                    }
                }
                objects.Remove(package.Key);
                package.ReplyChannel.Send(ReturnCode.Success);

            }
            else
            {
                package.ReplyChannel.Send(ReturnCode.ObjectNotFound);
            }
        }

        private void UnlockObject(UnlockObjectEntry uoe)
        {
            if (objects.ContainsKey(uoe.Key))
            {
                if (objects[uoe.Key].WaitingProcesses.Count > 0)
                {
                    Log.Instance.AddLogEntry("ObjectManager", "Removed from waiting queue");
                    Channel<object> channel = objects[uoe.Key].WaitingProcesses.Dequeue();
                    channel.Send(objects[uoe.Key].ObjectLock);
                }
                else
                {
                    objects[uoe.Key].Busy = false;
                }

            }
        }

        private void LockObject(LockObjectEntry loe)
        {
            if (objects.ContainsKey(loe.Key))
            {
                if (objects[loe.Key].Busy || objects[loe.Key].Negotiating)
                {
                    Log.Instance.AddLogEntry("ObjectManager", "Added to waiting queue");
                    objects[loe.Key].WaitingProcesses.Enqueue(loe.ReplyChannel);
                }
                else
                {
                    objects[loe.Key].Busy = true;
                    loe.ReplyChannel.Send(objects[loe.Key].ObjectLock);
                }
            }
            else
            {
                loe.ReplyChannel.Send(ReturnCode.ObjectNotFound);
            }
        }

        private void createObject(ObjectCreate package)
        {
            Log.Instance.AddLogEntry("ObjectManager", package.Key);
            if (objects.ContainsKey(package.Key))
            {
                if (package.ReplyChannel != null)
                {
                    package.ReplyChannel.Send(ReturnCode.ObjectAlreadyExist);
                }
                Log.Instance.AddLogEntry("ObjectManager", "Heureka - create was called for an object that already exist: " + package.Key);
            }
            else
            {
                //create object
                ObjectLock ol = new ObjectLock();
                ol.Policy = package.Action.Policy;
                ol.Coordinator = false;
                ol.ObjectInformation = new ObjectInformation(0L, package.Action.Objects.Count, 0L);

                ObjectEntry oe = new ObjectEntry();

                oe.Busy = true;
                oe.Negotiating = true;
                oe.ObjectLock = ol;

                if (package.ReplyChannel != null) //negotiation is already undertaken
                {
                    ulong bullyNumber = RandomGenerator.ULong;
                    oe.BullyNumber = bullyNumber;
                    package.ReplyChannel.Send(ReturnCode.Success);
                    package.ReplyChannel.Send(bullyNumber);
                }

                objects.Add(package.Key, oe);
            }
        }

        private void addObjectsForDevice(IStorageDevice device)
        {
            ObjectDevice oDevice = new ObjectDevice(device);

            Dictionary<string, ObjectLock> objs = oDevice.GetObjects();

            foreach (string key in objs.Keys)
            {
                if (objects.ContainsKey(key))
                {
                    if (objects[key].ObjectLock.ObjectInformation.Version < objs[key].ObjectInformation.Version)
                    {
                        objects[key].ObjectLock = objs[key];
                    }
                    Log.Instance.AddLogEntry("ObjectManager", key + " replaced by newer version");
                }
                else
                {
                    ObjectEntry oe = new ObjectEntry();
                    oe.ObjectLock = objs[key];
                    objects.Add(key, oe);
                    Log.Instance.AddLogEntry("ObjectManager", key + " added");
                }

            }

            oDevice.Benchmark();
            objectDevices.Add(oDevice);
            objectDevices.Sort();
        }
    }

    class ObjectEntry
    {
        public bool Busy = false;

        public ulong BullyNumber = 0L;
        public bool Negotiating = false;
        public Queue<Channel<object>> WaitingProcesses = new Queue<Channel<object>>();

        public ObjectLock ObjectLock = null;
    }

    class ObjectLock
    {
        private object mutex = new object();

        private Policy policy = null;
        public Policy Policy
        {
            get
            {
                Policy result;
                lock (mutex)
                {
                    result = policy.Copy();
                }
                return result;
            }
            set
            {
                lock (mutex)
                {
                    policy = value;
                }
            }
        }

        private ObjectInformation objectInformation = null;
        public ObjectInformation ObjectInformation
        {
            get
            {
                ObjectInformation result = null;
                lock (mutex)
                {
                    if (objectInformation != null)
                        result = objectInformation.Copy();
                }
                return result;
            }
            set
            {
                if (value != null && objectInformation != null)
                    Log.Instance.AddLogEntry("ObjectInformation", "Changed from " + objectInformation.ToString() + " to " + value.ToString());

                lock (mutex)
                {
                    objectInformation = value;
                }
            }
        }

        private ObjectDevice device = null;
        public ObjectDevice Device
        {
            get
            {
                lock (mutex)
                {
                    return device;
                }
            }
            set
            {
                lock (mutex)
                {
                    device = value;
                }
            }
        }

        public List<PeerContainer> ReplicatingPeers = new List<PeerContainer>();

        public bool Coordinator = false;

        private PeerContainer externalCoordinator = null;
        public PeerContainer ExternalCoordinator
        {
            get
            {
                lock (mutex)
                {
                    return externalCoordinator;
                }
            }
            set
            {
                lock (mutex)
                {
                    externalCoordinator = value;
                }
            }
        }
    }

    [Serializable]
    class SimpleObjectEntry
    {
        public Policy Policy = null;
        public string key = "";
        public bool Coordinator = true;
    }
}
